package cn.com.csg.smartenergy.mqtt;

import cn.com.csg.smartenergy.common.Constant;
import cn.com.csg.smartenergy.common.DateUtils;
//import com.alibaba.druid.util.StringUtils;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.util.*;

import static cn.com.csg.smartenergy.common.Constant.FATHER_NODE_ID;

@Slf4j
@Data
@Component
public class DeviceDataReportHandler {

//    @Autowired
//    private IotDeviceService deviceService;
//
//    @Autowired
//    private IotDeviceMessageService deviceMessageService;
//
//    @Autowired
//    private IotDeviceDataService deviceDataService;
//
//    @Autowired
//    private PowerFaulalarmService faulalarmService;
//
//    @Autowired
//    private PowerTopologyNodeReportService powerTopologyNodeReportService;
//
//    @Autowired
//    private PowerLineLossRecordService lineLossRecordService;
//
//    @Autowired
//    private PowerLineLossConfigService lineLossConfigService;
//
//    @Autowired
//    private RedisService redisService;

    private AbstractMqttClient client;

    @Value("${real.time.data.prefix}")
    private String prefix;

    /**
     * 是否开启线损计算
     */
    @Value("${config.lineloss.open}")
    private boolean lineLossFlag;

    /**
     * 冻结时间字段
     */
    @Value("${config.lineloss.freezeProp}")
    private String freezeProp;

    /**
     * 实时数据
     */
    public static final String REAL_DATA = "/device/realData/";

    /**
     * 告警
     */
    public static final String FAULALARM_DATA = "/device/faulalarmData/";

    private static final Set<String> METHODS = new HashSet<>(Arrays.asList("parameter_Get", "analog_Get", "discrete_Get", "accumulator_Get"));

    /**
     * 消息接收回调异步方法
     * APP主动上报数据topic: @app/terminal/dataReport/@manufacturerId/@manufacturerName/@deviceType/@model/@nodeId
     * 末端设备上线通知接口topic: esdk/notify/event/terminal/status/@manufacturerId/@manufacturerName/@deviceType/@model/@protocolType/@nodeId
     * 末端设备上报紧急事件接口topic: esdk/notify/event/terminal/emergency/@manufacturerId/@manufacturerName/@deviceType/@model /@nodeId
     * 业务APP响应命令topic：@app/terminal/cmdRsp/@manufacturerId/@manufacturerName/@deviceType/@model/@nodeId
     *
     * @param topic   主题
     * @param message 消息
     */
    @Async
    public void handleDataReportMessage(String topic, MqttMessage message, AbstractMqttClient client) throws MqttException {
//        this.client = client;
//        String msgPayload = new String(message.getPayload(), StandardCharsets.UTF_8);
//        log.info(String.format("设备上报数据：Topic: %s, Payload: %s", topic, msgPayload));
//        String[] topics = topic.split(Constant.FORWARD_SLASH);
//        Long deviceId;
//        String clientId = new StringBuilder()
//                .append(topics[3]).append(Constant.UNDER_LINE)
//                .append(topics[5]).append(Constant.UNDER_LINE)
//                .append(topics[6]).append(Constant.UNDER_LINE)
//                .append(topics[7]).toString();
//        if (DeviceMqttOnlineClient.DEVICE_IDS.containsKey(clientId)) {
//            deviceId = DeviceMqttOnlineClient.DEVICE_IDS.get(clientId);
//        } else {
//            // APP主动数据上报
//            IotDeviceEntity params = new IotDeviceEntity();
//            params.setManufacturerId(topics[3]);
//            params.setDeviceType(topics[5]);
//            params.setModel(topics[6]);
//            params.setNodeId(topics[7]);
//            IotDeviceEntity device = deviceService.getDeviceInfo(params);
//            if (device == null) {
//                log.info(String.format("数据库中找不到设备信息：ManufacturerId: %s, DeviceType: %s, Model: %s, NodeId: %s", topics[3], topics[5], topics[6], topics[7]));
//                return;
//            }
//            deviceId = device.getId();
//            DeviceMqttOnlineClient.DEVICE_IDS.put(clientId, deviceId);
//        }
//        // 数据上报或者命令请求
//        if (topic.contains("/terminal/dataReport/") || topic.contains("/terminal/cmdRsp/")) {
//            IotDeviceMessageEntity deviceMessageEntity = new IotDeviceMessageEntity();
//            deviceMessageEntity.setDeviceId(deviceId);
////            deviceMessageEntity.setCreateTime(new Date());
//            JSONObject payload = JSON.parseObject(msgPayload);
//            deviceMessageEntity.setMessageTime(payload.getDate(Constant.MSG_TIMESTAMP));
//            deviceMessageEntity.setMessage(msgPayload);
//            // 保存设备信息
////            log.info(String.format("保存设备上报数据：deviceId: %d", deviceId));
////            deviceService.saveDeviceMessage(deviceMessageEntity);
//            if (topic.contains("/terminal/dataReport/")) {
//                this.redisService.lRightPush(prefix.toUpperCase() + "_" + Constant.QUEUE_DEVICE_MESSAGE, JSON.toJSONString(deviceMessageEntity));
//            }
//            JSONObject body = payload.getJSONObject(Constant.MSG_BODY);
//            // 用于更新redis
//            Map<String, String> servicePropsMap = new HashMap<>();
//            if (body != null) {
//                JSONObject serviceProps = body.getJSONObject(Constant.MSG_SERVICE_PROPS);
//                if (serviceProps != null) {
//                    if (lineLossFlag) {
//                        // 保存线损数据
//                        this.saveLineLossRecord(deviceId, serviceProps);
//                    }
//                    List<IotDeviceDataEntity> deviceDataList = new ArrayList<>();
//                    Date messageTime = payload.getDate(Constant.MSG_TIMESTAMP);
//                    String serviceId = body.getString(Constant.MSG_SERVICE_ID);
//                    // yyyy-MM-dd HH:mm:ss 格式化时间
//                    String strMessageTime = DateUtils.formatTime(messageTime);
//                    String key = prefix + "_data_" + deviceId;
//                    JSONObject jsonServiceProps = null;
//                    // 如果上来的是遥信
//                    if (Constant.DISCRETE.equals(serviceId) || (Constant.TOPOLOGY.equals(serviceId) && null != serviceProps.get(FATHER_NODE_ID))) {
//                        Object dbServiceProps = redisService.hGet(key, serviceId);
//                        if (dbServiceProps != null) {
//                            jsonServiceProps = JSON.parseObject(dbServiceProps.toString());
//                        }
//                    }
//                    for (String property : serviceProps.keySet()) {
//                        IotDeviceDataEntity deviceData = new IotDeviceDataEntity();
//                        // 设备ID
//                        deviceData.setDeviceId(deviceId);
//                        // 消息ID
//                        deviceData.setMessageId(deviceMessageEntity.getId());
//                        // 服务
//                        deviceData.setServiceId(serviceId);
//                        // 服务的属性
//                        deviceData.setServiceProperty(property);
//                        // 上报的属性值
//                        String propertyValue = serviceProps.getString(property);
//                        // 属性值
//                        deviceData.setDataValue(propertyValue);
//                        // 上报时间
//                        deviceData.setMessageTime(messageTime);
//                        // 创建时间
////                        deviceData.setCreateTime(new Date());
//                        deviceDataList.add(deviceData);
//                        servicePropsMap.put(property, serviceProps.getString(property) + Constant.SEPARTOR + strMessageTime);
//
//                        // 如果上来的是拓扑图且是父节点nodeId
//                        if (Constant.TOPOLOGY.equals(serviceId) && FATHER_NODE_ID.equals(property)) {
//                            if (!StringUtils.isEmpty(propertyValue)) {
//                                // 如果redis有父节点nodeId的实时数据, 判断与redis中是否相同, 如果相同则不用处理, 除此之外的情况全部入队列
//                                if (!(jsonServiceProps != null && !StringUtils.isEmpty(jsonServiceProps.getString(property)) && propertyValue.equals(jsonServiceProps.getString(property).split(Constant.PARSE_SEPARTOR)[0]))) {
//                                    PowerTopologyNodeReportEntity topoNode = new PowerTopologyNodeReportEntity();
//                                    topoNode.setNodeId(topics[7]);
//                                    topoNode.setFatherNodeId(propertyValue);
//                                    // 添加到redis队列
//                                    this.redisService.lRightPush(prefix.toUpperCase() + "_" + Constant.QUEUE_TOPO_NODE, JSON.toJSONString(topoNode));
//                                }
//                            } else {
//                                log.info(String.format("拓扑识别：Topic: %s, Payload: %s, 父节点nodeId不能为空", topic, msgPayload));
//                            }
//                        }
//
//                        if (Constant.DISCRETE.equals(serviceId)) {
//                            boolean hasFaulalarm = false;
//                            // 如果缓存中没有该遥信的状态信息
//                            if (jsonServiceProps == null || StringUtils.isEmpty(jsonServiceProps.getString(property))) {
//                                // 遥信状态 1：异常 0：正常
//                                Integer state = serviceProps.getInteger(property);
//                                if (state == 1) {
//                                    hasFaulalarm = true;
//                                }
//                            } else {
//                                // 格式：遥信值|时间戳
//                                String[] propertyValues = jsonServiceProps.getString(property).split(Constant.PARSE_SEPARTOR);
//                                // 如果遥信变位了
//                                if (propertyValues.length == 2 && !propertyValues[0].equals(propertyValue)) {
//                                    hasFaulalarm = true;
//
//                                }
//                            }
//                            // 如果有告警
//                            if (hasFaulalarm) {
//                                PowerFaulalarmEntity faulalarmEntity = new PowerFaulalarmEntity();
//                                // 设备ID
//                                faulalarmEntity.setDeviceId(deviceId);
//                                // 遥信类型
//                                faulalarmEntity.setType(property);
//                                // 遥信状态
//                                faulalarmEntity.setState(Integer.valueOf(propertyValue));
//                                // 处理状态 默认为0
//                                faulalarmEntity.setStatus(0);
//                                // 上报时间
//                                faulalarmEntity.setMessageTime(messageTime);
//                                // websocket 推送
//                                Map<Object, Object> faulalarmData = new HashMap<>();
//                                faulalarmData.put("deviceId", deviceId);
//                                faulalarmData.put("type", property);
//                                faulalarmData.put("state", Integer.valueOf(propertyValue));
//                                faulalarmData.put("status", 0);
//                                faulalarmData.put("messageTime", messageTime);
//                                String faulalarmDataTopic = prefix + FAULALARM_DATA + deviceId;
//                                MqttMessage faulalarmDataMessage = new MqttMessage();
//                                faulalarmDataMessage.setQos(1);
//                                String faulalarmDataStr = JSONObject.toJSONString(faulalarmData);
//                                faulalarmDataMessage.setPayload(faulalarmDataStr.getBytes());
//                                this.client.publish(faulalarmDataTopic, faulalarmDataMessage);
//                                // 添加到redis队列
//                                this.redisService.lRightPush(prefix.toUpperCase() + "_" + Constant.QUEUE_ALARM, JSON.toJSONString(faulalarmEntity));
//                            }
//                        }
//                    }
////                    log.info(String.format("批量保存设备上报解析后数据：message: %d", deviceMessageEntity.getId()));
//                    // 实时数据推送
//                    Map<Object, Object> realData = new HashMap<>();
//                    realData.put(serviceId, JSON.toJSONString(servicePropsMap));
//                    realData.put(Constant.DEVICE_ID, deviceId);
//                    // 从缓存中获取 mqtt client
//                    String realDataTopic = prefix + REAL_DATA + deviceId;
//                    MqttMessage realDataMessage = new MqttMessage();
//                    realDataMessage.setQos(1);
//                    String realDataStr = JSONObject.toJSONString(realData);
////                    log.info(String.format("实时数据推送数据：%s", realDataStr));
//                    realDataMessage.setPayload(realDataStr.getBytes());
//                    this.client.publish(realDataTopic, realDataMessage);
////                    log.info(String.format("设备最新上报数据：serviceId: %s  serviceProperties: %s", serviceId, JSON.toJSONString(serviceProps)));
////                    String dataTableName = TableNameUtils.getTableName(Constant.IOT_DEVICE_DATA, messageTime);
////                    deviceService.saveBatchDeviceData(dataTableName, deviceDataList);
////                    String key = Constant.DATA + Constant.UNDER_LINE + deviceId;
//                    this.mergeRedisData(servicePropsMap, serviceId, key);
//                }
//            }
//
//        /*} else if (topic.contains("/terminal/cmdRsp/")) {
//            // 获取消息负荷
//            JSONObject payload = JSON.parseObject(msgPayload);
//            JSONObject body = payload.getJSONObject(Constant.MSG_BODY);
//            if (body != null) {
//                // 获取指令名称
//                String method = body.getString(Constant.CMD_METHOD);
//                if (!StringUtils.isEmpty(method) && METHODS.contains(method)) {
//                    JSONObject serviceProps = body.getJSONObject(Constant.MSG_SERVICE_PROPS);
//                    if (serviceProps != null) {
//                        Date messageTime = payload.getDate(Constant.MSG_TIMESTAMP);
//                        String serviceId = body.getString(Constant.MSG_SERVICE_ID);
//                        // yyyy-MM-dd HH:mm:ss 格式化时间
//                        String strMessageTime = DateUtils.formatTime(messageTime);
//                        Map<String, String> servicePropsMap = new HashMap<>();
//                        for (String property : serviceProps.keySet()) {
//                            servicePropsMap.put(property, serviceProps.getString(property) + Constant.SEPARTOR + strMessageTime);
//                        }
//                        String key = prefix + "_data_" + deviceId;
//                        log.info(String.format("设备响应命令上报数据：serviceId: %s  serviceProperties: %s", serviceId, JSON.toJSONString(serviceProps)));
//                        // 如果redis 包含key
//                        this.mergeRedisData(servicePropsMap, serviceId, key);
//                    }
//                }
//            }*/
//
//        } else if (topic.contains("esdk/notify/event/terminal/status/")) {
//            // 末端设备上线通知接口，暂时没有相关逻辑
//
//        } else if (topic.contains("esdk/notify/event/terminal/")) {
//            // 末端设备上报紧急事件接口，暂时没有相关逻辑
//
//        } else {
//            // 未知Topic
//            log.info(String.format("未知主题：Topic: %s, Payload: %s", topic, msgPayload));
//        }
    }


    /**
     * 保存线损数据
     *
     * @param deviceId     设备ID
     * @param serviceProps 服务属性
     */
    public void saveLineLossRecord(Long deviceId, JSONObject serviceProps) {
//        String key = prefix + "_line_loss";
//        Object config = redisService.hGet(key, deviceId.toString());
//        String property;
//        if (config == null) {
//            PowerLineLossConfigEntity configEntity = lineLossConfigService.getBaseMapper()
//                    .selectOne(new QueryWrapper<PowerLineLossConfigEntity>().eq("device_id", deviceId));
//            if (configEntity == null) {
//                return;
//            }
//            property = configEntity.getProperty();
//            redisService.hPut(key, deviceId.toString(), JSONObject.toJSONString(configEntity));
//        } else {
//            JSONObject configInfo = JSONObject.parseObject(config.toString());
//            property = configInfo.getString("property");
//        }
//        BigDecimal freezePower = serviceProps.getBigDecimal(property).setScale(4, BigDecimal.ROUND_HALF_UP);
//        if (freezePower == null) {
//            return;
//        }
//        // 获取冻结时间
//        Date freezeTime = serviceProps.getDate(freezeProp);
//        if (freezeTime == null) {
//            return;
//        }
//        PowerLineLossRecordEntity lineLossRecordEntity = new PowerLineLossRecordEntity();
//        lineLossRecordEntity.setDeviceId(deviceId);
//        lineLossRecordEntity.setFreezePower(freezePower);
//        lineLossRecordEntity.setFreezeTime(freezeTime);
//        // 添加到redis队列
//        this.redisService.lRightPush(prefix.toUpperCase() + "_" + Constant.QUEUE_LINE_LOSS, JSON.toJSONString(lineLossRecordEntity));
    }

    /**
     * 批量处理设备上报数据
     */
    @Async
    public void handleDeviceMessageQueue() {
        log.info("开始处理设备上报数据");
        List<String> messageList = new ArrayList<>();
//        while (true) {
//            try {
//                // 如果messageList不为空, 说明上一次取到值, 则不等待直接进下一轮, 避免高峰期的多余等待
//                if (null == messageList || messageList.isEmpty()) {
//                    Thread.sleep(100);
//                }
//                messageList = redisService.lRange(prefix.toUpperCase() + "_" + Constant.QUEUE_DEVICE_MESSAGE, 0, 1999);
//                if (messageList != null && !messageList.isEmpty()) {
//                    redisService.lTrim(prefix.toUpperCase() + "_" + Constant.QUEUE_DEVICE_MESSAGE, messageList.size(), -1);
//                    List<IotDeviceMessageEntity> messages = new ArrayList<>();
//                    for (String message : messageList) {
//                        IotDeviceMessageEntity messageEntity = JSON.parseObject(message, IotDeviceMessageEntity.class);
//                        messages.add(messageEntity);
//                    }
//                    this.deviceMessageService.getBaseMapper().saveBatchMessage(messages);
//                    // log.info(String.format("批量插入设备报文信息：个数: %d", messages.size()));
//                    List<IotDeviceDataEntity> deviceDataList = new ArrayList<>();
//                    // 遍历所有的上报数据
//                    for (IotDeviceMessageEntity message : messages) {
//                        // 消息负荷
//                        JSONObject payload = JSON.parseObject(message.getMessage());
//                        JSONObject body = payload.getJSONObject(Constant.MSG_BODY);
//                        // 用于更新redis
//                        if (body != null) {
//                            // 解析报文中的属性信息
//                            JSONObject serviceProps = body.getJSONObject(Constant.MSG_SERVICE_PROPS);
//                            if (serviceProps != null) {
//                                Date messageTime = payload.getDate(Constant.MSG_TIMESTAMP);
//                                String serviceId = body.getString(Constant.MSG_SERVICE_ID);
//                                for (String property : serviceProps.keySet()) {
//                                    IotDeviceDataEntity deviceData = new IotDeviceDataEntity();
//                                    // 设备ID
//                                    deviceData.setDeviceId(message.getDeviceId());
//                                    // 消息ID
//                                    deviceData.setMessageId(message.getId());
//                                    // 服务
//                                    deviceData.setServiceId(serviceId);
//                                    // 服务的属性
//                                    deviceData.setServiceProperty(property);
//                                    // 属性值
//                                    deviceData.setDataValue(serviceProps.getString(property));
//                                    // 上报时间
//                                    deviceData.setMessageTime(messageTime);
//                                    // 创建时间
//                                    // deviceData.setCreateTime(new Date());
//                                    deviceDataList.add(deviceData);
//                                }
//                            }
//                        }
//                        // 由于如果一次批量处理记录过多，则速度同样很慢，所以分批进行批处理
//                        if (deviceDataList.size() >= 1000) {
//                            this.deviceDataService.getBaseMapper().saveBatchDeviceData(deviceDataList);
//                            // log.info(String.format("批量插入设备报文解析后信息：个数: %d", deviceDataList.size()));
//                            deviceDataList.clear();
//                        }
//                    }
//                    if (deviceDataList.size() > 0) {
//                        this.deviceDataService.getBaseMapper().saveBatchDeviceData(deviceDataList);
//                        // log.info(String.format("批量插入设备报文解析后信息：个数: %d", deviceDataList.size()));
//                        deviceDataList.clear();
//                    }
//
//                }
//                // 告警队列
//                List<String> alarmList = redisService.lRange(prefix.toUpperCase() + "_" + Constant.QUEUE_ALARM, 0, 1999);
//                if (alarmList != null && !alarmList.isEmpty()) {
//                    redisService.lTrim(prefix.toUpperCase() + "_" + Constant.QUEUE_ALARM, alarmList.size(), -1);
//                    List<PowerFaulalarmEntity> faulalarmEntities = new ArrayList<>();
//                    for (String alarm : alarmList) {
//                        PowerFaulalarmEntity faulalarmEntity = JSON.parseObject(alarm, PowerFaulalarmEntity.class);
//                        faulalarmEntities.add(faulalarmEntity);
//                    }
//                    this.faulalarmService.saveBatch(faulalarmEntities);
//                }
//                // 拓扑识别队列
//                List<String> topoNodeList = redisService.lRange(prefix.toUpperCase() + "_" + Constant.QUEUE_TOPO_NODE, 0, 1999);
//                if (topoNodeList != null && !topoNodeList.isEmpty()) {
//                    redisService.lTrim(prefix.toUpperCase() + "_" + Constant.QUEUE_TOPO_NODE, topoNodeList.size(), -1);
//                    List<PowerTopologyNodeReportEntity> topoNodes = new ArrayList<>();
//                    for (String topoNodeStr : topoNodeList) {
//                        PowerTopologyNodeReportEntity topoNode = JSON.parseObject(topoNodeStr, PowerTopologyNodeReportEntity.class);
//                        topoNodes.add(topoNode);
//                    }
//                    this.powerTopologyNodeReportService.saveOrUpdateBatch(topoNodes);
//                }
//                if (lineLossFlag) {
//                    // 线损值入库
//                    List<String> lineLossRecords = redisService.lRange(prefix.toUpperCase() + "_" + Constant.QUEUE_LINE_LOSS, 0, 1999);
//                    if (lineLossRecords != null && !lineLossRecords.isEmpty()) {
//                        redisService.lTrim(prefix.toUpperCase() + "_" + Constant.QUEUE_LINE_LOSS, lineLossRecords.size(), -1);
//                        List<PowerLineLossRecordEntity> records = new ArrayList<>();
//                        for (String lineLossStr : lineLossRecords) {
//                            PowerLineLossRecordEntity lineLossEntity = JSON.parseObject(lineLossStr, PowerLineLossRecordEntity.class);
//                            records.add(lineLossEntity);
//                        }
//                        this.lineLossRecordService.saveOrUpdateBatch(records);
//                    }
//                }
//
//            } catch (Exception e) {
//                log.error(String.format("设备数据入库时出现异常"), e);
//            }
//        }
    }

    /**
     * 将最新上报的数据合并到redis中
     *
     * @param servicePropsMap 属性
     * @param serviceId       服务ID
     * @param key             key
     */
    private void mergeRedisData(Map<String, String> servicePropsMap, String serviceId, String key) {
        System.out.println("mergeRedisData");
//        if (this.redisService.hasKey(key)) {
//            Map<Object, Object> dbData = this.redisService.hGetAll(key);
////            log.info(String.format("Redis保存设备数据：dbData: %s", JSON.toJSONString(dbData)));
//            // redis 获取的数据
//            Object dbServiceProps = dbData.get(serviceId);
//            // 如果 serviceId 不存在
//            if (dbServiceProps == null) {
//                // 将最新上报的ServiceProps放到redis
//                dbData.put(serviceId, JSON.toJSONString(servicePropsMap));
//            } else {
//                Map<Object, Object> parseServiceProps = JSON.parseObject(dbServiceProps.toString(), Map.class);
//                parseServiceProps.putAll(servicePropsMap);
//                dbData.put(serviceId, JSON.toJSONString(parseServiceProps));
//            }
//            // 将新上来的数据合并到redis中
////            log.info(String.format("合并后的设备数据：dbData: %s", JSON.toJSONString(dbData)));
//            this.redisService.hPutAll(key, dbData);
//        } else {
//            Map<Object, Object> newData = new HashMap<>();
//            // 按照结构构造最新的数据
//            newData.put(serviceId, JSON.toJSONString(servicePropsMap));
////            log.info(String.format("合并后的设备数据：dbData: %s", JSON.toJSONString(newData)));
//            this.redisService.hPutAll(key, newData);
//        }
    }
}
